package com.wlz.confirm;

import com.rabbitmq.client.*;
import com.wlz.utils.RabbitUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 *  主题（通配符） 模式 -- 生产者
 *  消息确认回退 confirm&return
 * @author wlz
 * @date 2022-06-27  10:38 下午
 */
@Slf4j
public class WeatherBureau {

    public static Map area = new LinkedHashMap<String,String>();

    static {
        area.put("china.hunan.changsha.20201127", "中国湖南长沙20201127天气数据");
        area.put("china.hubei.wuhan.20201127", "中国湖北武汉20201127天气数据");
        area.put("china.hunan.zhuzhou.20201127", "中国湖南株洲20201127天气数据");
        area.put("us.cal.lsj.20201127", "美国加州洛杉矶20201127天气数据");

        area.put("china.hebei.shijiazhuang.20201128", "中国河北石家庄20201128天气数据");
        area.put("china.hubei.wuhan.20201128", "中国湖北武汉20201128天气数据");
        area.put("china.henan.zhengzhou.20201128", "中国河南郑州20201128天气数据");
        area.put("us.cal.lsj.20201128", "美国加州洛杉矶20201128天气数据");
    }

    public static void main(String[] args) throws Exception{

        Connection connection = RabbitUtils.getConnection();
        // 控制台输入
        Channel channel = connection.createChannel();


        // 开启confirm 监听模式
        channel.confirmSelect();

        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                log.info("消息已被Broker 接收， tag: {}", l);
            }

            @Override
            public void handleNack(long l, boolean b) throws IOException {
                log.info("消息已被Broker 拒收， tag: {}", l);
            }
        });
        channel.addReturnListener(new ReturnCallback() {
            @Override
            public void handle(Return aReturn) {
                log.info("==============================");
                log.info("Return编码: {}, -Rrturn描述: {}", aReturn.getReplyCode(), aReturn.getReplyText());
                log.info("交换机: {}, -路由key: {}", aReturn.getExchange(), aReturn.getRoutingKey());
                log.info("Retrun主题: {}", new String(aReturn.getBody()));
                log.info("==============================");
            }
        });


        String exchangeName = "wlz_weather_topic_confirm";
        Iterator iterator = area.entrySet().iterator();
        while (iterator.hasNext()) {

            Map.Entry<String,String> entry = (Map.Entry<String, String>) iterator.next();

            // 第一个参数 为 交换机名字 , 第二个参数作为 消息的routing key,
            // 第三个参数为：mandatory true代表如果消息无法正常投递则return回生产者，如果false，则直接将消息放弃。
            // 其他参数和之前一样， (其中 交换机必须先创建，不然会报错， 选择 topic 类型)
            channel.basicPublish(exchangeName, entry.getKey(), true,null, entry.getValue().getBytes());
        }

        log.info("发布定义模式 交换机为: {} , 发送消息成功  ",exchangeName);

//        channel.close();
//        connection.close();


    }
}
